package dataflow.core;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.eclipse.emf.common.util.EList;

import dataflowScheme.ConnectionType;
import dataflowScheme.Port;
import dataflowScheme.PortSet;
import dataflowScheme.SynchroPort;

public abstract class PE implements IPE {

	static protected Logger log = Logger.getLogger(PE.class);
	
	/** source class on which processing will be called */
	protected Class<?> source;
	/** source instance*/
	protected Object sourceInstance;
	/** map of port set <-> method which will be called on source */
//	private Map<Integer, Method> runMethods = new HashMap<Integer, Method>();
	protected Method runMethod;
	/** appropriate Processing entity instance from input EMF scheme */
	protected dataflowScheme.PE ePe;
	
	/** map of port name <-> method of 'source' used to set data for port */
	protected Map<String, Method> methods = new HashMap<String, Method>();
	/** map of port name <-> data which flow to this task */
	protected Map<String, Object> values = new HashMap<String, Object>();
//	/** map of port <-> number of port set */
//	protected Map<String, Integer> portsMap = new HashMap<String, Integer>();
	/** map of port name <-> Port instance */
	protected Map<String, Port> portsMap = new HashMap<String, Port>();
	/** map of port set <-> current amount of data (when = 0 means need to run source) */
	protected Map<Integer, Integer> portSetRate = new HashMap<Integer, Integer>();
//	/** map of port set <-> amount of data necessary to execute task */
//	protected Map<Integer, Integer> portSetDefaultRate = new HashMap<Integer, Integer>();
	/** map of port name <-> connection anchor where data are coming 'from' or 'to' */
	protected Map<String, ConnectionAnchor> oppossitePort = new HashMap<String, ConnectionAnchor>();
	/** port name <-> produced data token */
	protected Map<String, Object> producedData = new HashMap<String, Object>();
	/** port name <-> signals to send */
	protected List<String> signals2send = new ArrayList<String>();
	
	protected Map<String, ConnectionAnchor> failedAttempts = new HashMap<String, ConnectionAnchor>();
	
	protected final LinkedBlockingQueue<IEvent> eventQueue;
	protected final LinkedBlockingQueue<ExecutionEvent> executionQueue;
	
	protected static Class<?>[] setParams = new Class<?>[1];
	protected static Class<?>[] getParams = new Class<?>[0];
	
	static{
		setParams[0] = java.lang.Object.class;
	}
	
	public PE(dataflowScheme.PE ePe, LinkedBlockingQueue<IEvent> queue, LinkedBlockingQueue<ExecutionEvent> executionQueue){
		this.ePe = ePe;
		this.eventQueue = queue;
		this.executionQueue = executionQueue;
	}
	
	public boolean init() {//throws Exception{
		String className = getClassName(ePe.getSource());
		Class<?>[] params = new Class<?>[1];
		params[0] = java.lang.Integer.class;
		boolean rv;
		
		try{
			source = Class.forName(className);
			runMethod = source.getMethod("run", params);

			rv = processMethods(ePe.getPortSet());
			
		}catch(ClassNotFoundException cnfe){
			log.error("PE '" + ePe.getName() + "' init interrupted - no class found", cnfe);
			return false;
		} 
		catch(NoSuchMethodException nsme){
			log.error("PE '" + ePe.getName() + "' init interrupted - no 'run' method found", nsme);
			return false;
		}
		return rv;
		
	}
	
	protected String getClassName(String resourceSource){
		String[] parts = resourceSource.split("/"); // /loadResource/model-src/dataflow/example/Worker2.java
		StringBuilder className = new StringBuilder();
		for(int i = 3; i < parts.length - 1; i++){
			className.append(parts[i]);
			className.append(".");
		}
		String fileName = parts[parts.length-1];
		className.append(fileName.split("[.]")[0]);
		
		log.debug(className.toString());
		
		return className.toString();
	}
	
	protected boolean processMethods(EList<PortSet> portsets){
		
		boolean returnValue = true;
		
		for(PortSet ps : portsets){
			EList<Port> ports = ps.getPort();
			portSetRate.put(ps.getId(), ports.size());
//			portSetDefaultRate.put(portsetNumber, ports.size());
			
			for(Port p : ports){
				String portName = p.getName();
				portsMap.put(portName, p);
				
				if(p instanceof SynchroPort){
//					portsMap.put(portName, portsetNumber);
					continue;
				}
				
				if(methods.containsKey(portName)){
					log.error("Port name '" + portName + "' is duplicated.");
					returnValue = false;
					continue;
				}
				boolean in = p.getType().equals(ConnectionType.IN);
				String methodPrefix = in?"set":"get";
				String methodName = methodPrefix + portName;
				try{
					Method m = source.getMethod(methodName, in?setParams:getParams);
					methods.put(portName, m);
//					portsMap.put(portName, portsetNumber);
				} catch (NoSuchMethodException nsme){
					log.error("Failed while looking for method '" + methodName + "' of '" + ePe.getName() + "'", nsme);
					returnValue = false;
				}
			}
		}
		
		return returnValue;
	}

	
	protected void sendResult(){
		Map<String, Object> producedDataCopy = new HashMap<String, Object>();
		producedDataCopy.putAll(producedData);
		
		for(Map.Entry<String, Object> e : producedDataCopy.entrySet()){
			sendData(e.getValue(), e.getKey());
		}

		List<String> signals2sendCopy = new ArrayList<String>();
		signals2sendCopy.addAll(signals2send);
		
		Object[] signalPorts = signals2sendCopy.toArray();
		String stopperPort = null;
		for(int i = 0; i < signalPorts.length; i++){
			String port = (String)signalPorts[i];
			if(oppossitePort.get(port).pe instanceof Stopper){
				stopperPort = port;
			} else {
				sendSignal(port);
			}
		}
		if(stopperPort != null){
			sendSignal(stopperPort);
		}
	}
	
	protected void sendData(Object data, String portName) {
		ConnectionAnchor in = oppossitePort.get(portName);
		ConnectionAnchor out = new ConnectionAnchor(this, portName);
		
		try{
			eventQueue.put(new DataEvent(out, data, in));
		} catch(InterruptedException ie){
			log.error("Trying to send data event when event queue is already dead.", ie);
		}
		
//		if(anchor.pe.receiveData(data, anchor.portName)){
//			//received successfully
//			producedData.remove(portName);
//			if(producedData.isEmpty() && signals2send.isEmpty()){
//				cleanForNextIteration();
//			}
//		}
	}
	
	protected void sendSignal(String portName){
		ConnectionAnchor in = oppossitePort.get(portName);
		ConnectionAnchor out = new ConnectionAnchor(this, portName);
		
		try{
			eventQueue.put(new SynchroEvent(out, in));
		} catch(InterruptedException ie){
			log.error("Trying to send synchro event when event queue is already dead.", ie);
		}
		
//		if(anchor.pe.receiveSignal(anchor.portName)){
//			//received successfully
//			signals2send.remove(portName);
//			if(producedData.isEmpty() && signals2send.isEmpty()){
//				cleanForNextIteration();
//			}
//		}
	}

	protected void cleanForNextIteration(){
		log.debug("cleaning for next iteration for '" + ePe.getName() + "'");
		endOfSourceRun();
	}
	
	@Override
	public boolean receiveData(Object data, String portName) {
		if(!portsMap.containsKey(portName)){
			log.error("Trying to set data to unknown port '" + portName + "'.");
			return false;
		}
		
		if(values.containsKey(portName)){
			//we have a token, therefore reject attempt of sender
			return false;
		}
		
		
		values.put(portName, data);
		int portsetId = ((PortSet)portsMap.get(portName).eContainer()).getId(); 
		int num = portSetRate.get(portsetId) - 1;
		
		if(num == 0){
			//time to make next iteration
//			portSetRate.put(portsetNumber, portSetDefaultRate.get(portsetNumber));
			invokeNextIteration(portsetId);
		}else{
			portSetRate.put(portsetId, num);
		}
		return true;
	}

	protected abstract void invokeNextIteration(int portSet);
	
	@Override
	public void receiveResponse(EventMsg event, String portName) {
		switch(event){
		case freePort:
			if(producedData.containsKey(portName)){
				sendData(producedData.get(portName), portName);
			}
			break;
		case delivered_ok:
			if(producedData.containsKey(portName)){
				producedData.remove(portName);
			} else if(signals2send.contains(portName)){
				signals2send.remove(portName);
			}
			
			if(failedAttempts.containsKey(portName)){
				eventQueue.offer(new DataReleaseEvent(failedAttempts.get(portName)));
			}
			
			if(producedData.isEmpty() && signals2send.isEmpty()){
				cleanForNextIteration();
			}
			break;
		}
	}

	@Override
	public boolean receiveSignal(String portName) {
		if(!portsMap.containsKey(portName)){
			log.error("Trying to set data to unknown port '" + portName + "'.");
			return false;
		}
		
		if(values.containsKey(portName)){
			//we have a token, therefore reject attempt of sender
			return false;
		}
		
		int portsetNumber = ((PortSet)portsMap.get(portName).eContainer()).getId();
		int num = portSetRate.get(portsetNumber) - 1;
		
		if(num == 0){
			//time to make next iteration
//			portSetRate.put(portsetNumber, portSetDefaultRate.get(portsetNumber));
			invokeNextIteration(portsetNumber);
		}else{
			portSetRate.put(portsetNumber, num);
		}
		return true;
	}

	@Override
	public void addConnection(String portName,
			IConnectableEntity oppossiteEntity, String oppossitePortName) {
		oppossitePort.put(portName, new ConnectionAnchor(oppossiteEntity, oppossitePortName));
		
	}

	@Override
	public void receiveResult(int returnPortset) {
		boolean success = true;
		try{
			success = parseResult(returnPortset);
		} catch(Exception e){
			log.error("Error while parsing result of '" + ePe.getName() + "'.", e);
			success = false;
		}
		if(!success){
			cleanForNextIteration();
		}
		
		Integer nextPortSet = getNextPortset2invokeRun();
		if(nextPortSet != null){
			run(nextPortSet);
		}
		
	}
	
	protected boolean parseResult(int returnPortset){
		if(returnPortset != -1){
			PortSet ps = findPortSetByID(ePe, returnPortset);
			if(ps == null){
				log.error("Bad number of portset.");
				return false;
			}
		
			for (Port p : ps.getPort()){
				
				if(p instanceof SynchroPort){
					signals2send.add(p.getName());
					continue;
				}
				Method getMethod = methods.get(p.getName());
				try{
					Object value = getMethod.invoke(sourceInstance, new Object[0]);
					producedData.put(p.getName(), value);
				}catch(InvocationTargetException ite){
					log.error("Invocation of port '" + p.getName() + "' failed.", ite);
					return false;
				} catch (IllegalArgumentException e) {
					log.error("Method '" + getMethod.getName() + "' can't have any arguments.",e);
					return false;
				} catch (IllegalAccessException e) {
					log.error("Method '" + getMethod.getName() + "' have to be public.",e);
					return false;
				} 
				
			}
			
			sendResult();
		}
		return true;
	}
	
	protected abstract void endOfSourceRun();
	
	protected abstract Integer getNextPortset2invokeRun();
	
	public dataflowScheme.PE getPE(){
		return ePe;
	}

	protected abstract void run(int portset);
	
	public Class<?> getSourceClass(){
		return source;
	}
	
	public Map<String, Method> getPortMethods(){
		return methods;
	}
	
	public Map<String, Object> getPortValues(){
		return values;
	}
	
	public Method getRunMethod(){
		return runMethod;
	}
	
	public boolean saveAttempt(String portName, ConnectionAnchor fromPE){
		failedAttempts.put(portName, fromPE);
		return true;
	}
	
	static public PortSet findPortSetByID(dataflowScheme.PE pe, int id){
		for (PortSet ps : pe.getPortSet()){
			if(ps.getId() == id){
				return ps;
			}
		}
		return null;
	}
	
	public Object getSourceInstance(){
		return sourceInstance;
	}
	
	public void setSourceInstance(Object sourceInstance){
		this.sourceInstance = sourceInstance;
	}
}
